package com.hub.realtime.quartz.task;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.hub.realtime.common.config.RealTimeConfig;
import com.hub.realtime.common.core.domain.model.ClusterInfo;
import com.hub.realtime.common.core.redis.RedisCache;
import com.hub.realtime.common.enums.JobRunState;
import com.hub.realtime.common.model.checkpoint.CheckPointDetail;
import com.hub.realtime.common.model.checkpoint.CheckPointRest;
import com.hub.realtime.common.utils.hadoop.HadoopUtil;
import com.hub.realtime.common.utils.http.HttpUtils;
import com.hub.realtime.framework.config.SpringContextConfig;
import com.hub.realtime.quartz.model.JobAlertContent;
import com.hub.realtime.resource.domain.ResCheckpoints;
import com.hub.realtime.resource.domain.ResFlinkJob;
import com.hub.realtime.resource.service.IResApplicationService;
import com.hub.realtime.resource.service.IResCheckpointsService;
import com.hub.realtime.resource.service.IResFlinkJobService;
import com.hub.realtime.system.service.ISysConfigService;
import com.streamxhub.streamx.common.util.ThreadUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


@Component("sysTask")
@Slf4j
public class SysTask {

    @Autowired
    RestTemplate restTemplate;

    @Autowired
    IResFlinkJobService flinkJobService;

    @Autowired
    IResApplicationService applicationService;

    @Autowired
    IResCheckpointsService checkpointsService;


    @Autowired
    RedisCache redisCache;

    @Autowired
    ISysConfigService configService;


    public void doTriggerSavePointAuto() {
        flinkJobService.triggerSavePointOnTimer();
    }

    /**
     * 获取最新的检查点
     */
    public void getLatestCheckPoints() {
        try {
            LambdaQueryWrapper<ResFlinkJob> flinkJobLambdaQueryWrapper = new LambdaQueryWrapper<>();
            flinkJobLambdaQueryWrapper.eq(ResFlinkJob::getJobState, JobRunState.RUNNING.get())
                    .or()
                    .eq(ResFlinkJob::getJobState, JobRunState.RESTARTING.get());
            List<ResFlinkJob> resFlinkJobs = flinkJobService.getBaseMapper().selectList(flinkJobLambdaQueryWrapper);
            if (resFlinkJobs != null && !resFlinkJobs.isEmpty()) {
                resFlinkJobs.forEach(job -> {
                    if (job.getDeployModel().contains("yarn")) {
                        if (StringUtils.isNotEmpty(job.getYarnId()) && StringUtils.isNotEmpty(job.getFlinkJobId())) {
                            ClusterInfo clusterInfo = applicationService.getClusterInfoByAppId(job.getApplicationId());
                            String getCheckPointUrl = HadoopUtil.getRMWebAppURL(clusterInfo, true)
                                    .concat("/proxy/")
                                    .concat(job.getYarnId())
                                    .concat("/jobs/")
                                    .concat(job.getFlinkJobId())
                                    .concat("/checkpoints");
                            String res = HttpUtils.sendGet(getCheckPointUrl);
                            if (StringUtils.isNotEmpty(res)) {
                                CheckPointRest checkPointRest = JSON.parseObject(res, CheckPointRest.class);
                                if (checkPointRest != null && checkPointRest.getLatest() != null && checkPointRest.getLatest().getCompleted() != null) {
                                    CheckPointDetail completed = checkPointRest.getLatest().getCompleted();
                                    if ("COMPLETED".equalsIgnoreCase(completed.getStatus()) && !completed.getIs_savepoint()) {
                                        LambdaQueryWrapper<ResCheckpoints> checkpointsLambdaQueryWrapper = new LambdaQueryWrapper<>();
                                        checkpointsLambdaQueryWrapper.eq(ResCheckpoints::getJobId, job.getId());
                                        ResCheckpoints resCheckpoints = checkpointsService.getBaseMapper().selectOne(checkpointsLambdaQueryWrapper);
                                        if (resCheckpoints != null) {
                                            checkpointsService.getBaseMapper().deleteById(resCheckpoints);
                                        }
                                        ResCheckpoints checkpoints = new ResCheckpoints();
                                        checkpoints.setJobId(job.getId());
                                        checkpoints.setPath(completed.getExternal_path());
                                        checkpoints.setCreateTime(DateUtil.date(completed.getTrigger_timestamp()));
                                        checkpoints.setCreateBy(" ");
                                        checkpointsService.getBaseMapper().insert(checkpoints);
                                    }
                                }
                            }

                        }
                    }
                });
            }

        } catch (Exception ex) {

            System.out.println(ex.getMessage());

        }

    }


    /***
     * 处理重启（定时多久监控触发一次）
     * @param unit 重启间隔单位，h 表示小时；m 表示分钟; s表示秒
     * @param interval 间隔周期
     * @param times 重启超过的次数
     * 比如10分钟之内重启超过五次 m 10 5
     */
    public void handelJobRestart(String unit, Integer interval, Integer times) {

        String alertUrl = configService.selectConfigByKey("sys:alert:url");

        executorService.execute(() -> {
            if (!unit.equals("h") && !unit.equals("m") && !unit.equals("s")) {
                return;
            }
            List<ResFlinkJob> resFlinkJobs = flinkJobService.getBaseMapper().selectList(new LambdaQueryWrapper<>());
            if (resFlinkJobs != null && resFlinkJobs.size() > 0) {
                resFlinkJobs.forEach(job -> {
                    if (job.getDeployModel().contains("yarn") && StringUtils.isNotEmpty(job.getYarnId())) {
                        String redisKey = "FlinkJobRestart_" + job.getId();
                        flinkJobService.syncJobState(job);
                        Long redisData = redisCache.getCacheObject(redisKey);
                        if (job.getJobState() == JobRunState.FAILED.get() || job.getJobState() == JobRunState.FINISHED.get()
                                || job.getJobState() == JobRunState.CANCELED.get()) {
                            //直接告警,任务运行失败
                            if (redisData != null) {
                                redisCache.deleteObject(redisKey);
                            }

                            if (job.getJobState() == JobRunState.FAILED.get() && StringUtils.isNotEmpty(job.getFlinkJobId())
                                    && StringUtils.isNotEmpty(job.getFlinkJobId().trim())) {
                                //告警失败
                                String msg = "目前作业已经失败";
                                sendAlert(alertUrl, job, msg);
                            }

                        } else {
                            //计算重启
                            if (job.getJobState() == JobRunState.RESTARTING.get()) {
                                //查看10分钟之内是否重启了大于等于N次了
                                if (redisData == null) {
                                    redisCache.setCacheObject(redisKey, 1L);
                                } else {
                                    redisCache.setCacheObject(redisKey, redisData + 1L);
                                }

                                if (redisCache.getCacheObject(redisKey) != null) {
                                    if (unit.equals("m")) {
                                        redisCache.expire(redisKey, interval, TimeUnit.MINUTES);
                                    }
                                    if (unit.equals("s")) {
                                        redisCache.expire(redisKey, interval, TimeUnit.SECONDS);
                                    }
                                    if (unit.equals("h")) {
                                        redisCache.expire(redisKey, interval, TimeUnit.HOURS);
                                    }
                                    Long currentTimes = redisCache.getCacheObject(redisKey);
                                    if (currentTimes >= times) {
                                        String msg = "目前作业正在重启,重启次数：" + currentTimes;
                                        sendAlert(alertUrl, job, msg);
                                    }

                                }
                            }
                        }

                    }
                });
            }
        });
    }


    protected ExecutorService executorService = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors() * 2,
            200,
            60L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1024),
            ThreadUtils.threadFactory("realtime-flink-monitor"),
            new ThreadPoolExecutor.AbortPolicy()
    );


    private void sendAlert(String alertUrl, ResFlinkJob job, String msg) {
        String profile = SpringContextConfig.getActiveProfile();

        String env = "";
        if ("sit".equals(profile)) {
            env = "测试环境";
        }
        if ("pre".equals(profile)) {
            env = "预发环境";
        }
        if ("prod".equals(profile)) {
            env = "生产环境";
        }
        if ("dev".equals(profile)) {
            env = "开发环境";
        }
        JobAlertContent jobAlertContent = new JobAlertContent();
        StringBuilder sb = new StringBuilder();
        sb.append("\n\n======flink 任务告警提醒,告警信息如下：=======\n\n");
        sb.append(msg + "\n\n");
        sb.append("YarnID: " + job.getYarnId() + "\n\n");
        sb.append("JobID: " + job.getFlinkJobId() + "\n\n");
        sb.append("作业名称: " + job.getJobName() + "\n\n");
        sb.append("作业ID: " + job.getId() + "\n\n");
        sb.append("当前环境: " + env + "\n");
        sb.append("告警时间: " + DateUtil.date() + "\n");
        sb.append("=============================================");
        Map<String, Object> content = new HashMap<>();
        content.put("text", sb.toString());
        jobAlertContent.setContent(content);
        try {
            ResponseEntity<String> res = restTemplate.postForEntity(alertUrl, jobAlertContent, String.class);
        } catch (Exception ex) {
            log.error("发送告警失败：" + ex.getMessage());
            ex.printStackTrace();
        }


    }


    public void queryJobState() {
        executorService.execute(() -> {
            LambdaQueryWrapper<ResFlinkJob> resFlinkJobLambdaQueryWrapper = new LambdaQueryWrapper<>();
            List<ResFlinkJob> resFlinkJobs = flinkJobService.getBaseMapper().selectList(resFlinkJobLambdaQueryWrapper);
            if (resFlinkJobs != null) {
                resFlinkJobs.forEach(job -> {
                    flinkJobService.syncJobState(job);
                });
            }
        });

    }


}
